package com.aidex.web.controller.biz;

import com.alibaba.dashscope.app.Application;
import com.alibaba.dashscope.app.ApplicationParam;
import com.alibaba.dashscope.app.ApplicationResult;
import com.alibaba.dashscope.exception.ApiException;
import com.alibaba.dashscope.exception.InputRequiredException;
import com.alibaba.dashscope.exception.NoApiKeyException;
import io.reactivex.Flowable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

/**
 * @author caozhen
 * @ClassName AliTongYiQianWen
 * @description: 阿里-通义千问
 * @date 2024年01月22日
 * @version: 1.0
 */
@Slf4j
public class AliTongYiQianWen {

    public static void streamResponse(String question, SseEmitter emitter) throws ApiException, NoApiKeyException,
            InputRequiredException {
        ApplicationParam param = ApplicationParam.builder().apiKey("sk-277643488a6c4ba9893e427ee50655f8").appId(
                "f87bdbf37dfa4b68a26566fc6f9e22b9").prompt(question)
                // 增量输出
                .incrementalOutput(true).build();

        Application application = new Application();
        // .streamCall()：流式输出内容
        Flowable<ApplicationResult> result = application.streamCall(param);

        // 使用非阻塞方式订阅流式数据
        Disposable disposable = result.subscribeOn(Schedulers.io()) // 在IO线程池中执行
                .observeOn(Schedulers.computation()) // 在计算线程池中处理结果
                .subscribe(
                        // onNext: 每次接收到数据时立即处理
                        data -> {
                            try {
                                String content = data.getOutput().getText();
                                if (content != null && !content.isEmpty()) {
                                    // 立即发送SSE事件，不等待后续数据
                                    emitter.send(SseEmitter.event().name("message").data(content));
                                    log.info("实时发送SSE数据: {}", content);
                                }
                            } catch (IOException e) {
                                log.error("发送SSE数据失败", e);
                                emitter.completeWithError(e);
                            }
                        },
                        // onError: 处理错误
                        error -> {
                            log.error("流式处理异常", error);
                            try {
                                emitter.send(SseEmitter.event().name("error").data("处理异常: " + error.getMessage()));
                            } catch (IOException e) {
                                log.error("发送错误信息失败", e);
                            }
                            emitter.completeWithError(error);
                        },
                        // onComplete: 流式输出完成
                        () -> {
                            try {
                                emitter.send(SseEmitter.event().name("done").data("[DONE]"));
                                emitter.complete();
                                log.info("流式输出完成");
                            } catch (IOException e) {
                                log.error("发送完成信号失败", e);
                                emitter.completeWithError(e);
                            }
                        });

        // 当SSE连接断开时，取消订阅
        emitter.onCompletion(() -> {
            if (!disposable.isDisposed()) {
                disposable.dispose();
                log.info("SSE连接完成，已取消订阅");
            }
        });

        emitter.onTimeout(() -> {
            if (!disposable.isDisposed()) {
                disposable.dispose();
                log.warn("SSE连接超时，已取消订阅");
            }
        });

        emitter.onError(throwable -> {
            if (!disposable.isDisposed()) {
                disposable.dispose();
                log.error("SSE连接错误，已取消订阅", throwable);
            }
        });
    }

    public static void streamResponseRep(String question) throws ApiException, NoApiKeyException,
            InputRequiredException {
        ApplicationParam param = ApplicationParam.builder().apiKey("sk-277643488a6c4ba9893e427ee50655f8").appId(
                "f87bdbf37dfa4b68a26566fc6f9e22b9").prompt(question)
                // 增量输出
                .incrementalOutput(true).build();

        Application application = new Application();
        // .streamCall()：流式输出内容
        Flowable<ApplicationResult> result = application.streamCall(param);
        // ApplicationResult result = application.call(param);

        Disposable disposable = result.subscribeOn(Schedulers.io()) // 在IO线程池中执行
                .observeOn(Schedulers.computation()) // 在计算线程池中处理结果
                .subscribe(
                        // onNext: 每次接收到数据时立即处理
                        data -> {
                            try {
                                String content = data.getOutput().getText();
                                if (content != null && !content.isEmpty()) {
                                    System.out.println(content);
                                    // 立即发送SSE事件，不等待后续数据
                                    log.debug("实时发送SSE数据: {}", content);
                                }
                            } catch (Exception e) {
                                log.error("发送SSE数据失败", e);
                            }
                        },
                        // onError: 处理错误
                        error -> {
                            log.error("流式处理异常", error);
                            try {
                            } catch (Exception e) {
                                log.error("发送错误信息失败", e);
                            }
                        },
                        // onComplete: 流式输出完成
                        () -> {
                            try {
                                log.info("流式输出完成");
                            } catch (Exception e) {
                                log.error("发送完成信号失败", e);
                            }
                        });

    }

    // 保留原有方法用于测试
    public static void simpleMultiModalConversationCall(String question) throws ApiException, NoApiKeyException,
            InputRequiredException {
        ApplicationParam param = ApplicationParam.builder().apiKey("sk-277643488a6c4ba9893e427ee50655f8").appId(
                "f87bdbf37dfa4b68a26566fc6f9e22b9").prompt(question)
                // 增量输出
                .incrementalOutput(true).build();

        Application application = new Application();
        // .streamCall（）：流式输出内容
        Flowable<ApplicationResult> result = application.streamCall(param);
        result.blockingForEach(data -> {
            System.out.printf("%s\n", data.getOutput().getText());
        });
    }

    public static void main(String[] args) throws NoApiKeyException, InputRequiredException {
        try {
            simpleMultiModalConversationCall("新能源车1年 黑户,推荐产品");
        } catch (ApiException | NoApiKeyException | InputRequiredException e) {
            System.out.printf("Exception: %s", e.getMessage());
            System.out.println("请参考文档： `https://help.aliyun.com/zh/model-studio/developer-reference/error-code` ");
        }
        System.exit(0);
    }
}
